package cn.migu.music.hangupsms.cachemanage;

import cn.migu.music.common.utils.LogUtils;
import cn.migu.music.hangupsms.cachemanage.event.AbsCacheEvent;
import cn.migu.music.hangupsms.cachemanage.handler.AbsCacheHandler;
import cn.migu.music.hangupsms.cachemanage.handler.BusinessDbDataCacheServices;
import cn.migu.music.hangupsms.cachemanage.handler.SmsSendFrequencyCacheHandler;
import cn.migu.music.hangupsms.cachemanage.handler.SmsSendProvinceCacheHandler;
import cn.migu.music.hangupsms.cachemanage.handler.SmsSendTimeCacheHandler;
import cn.migu.music.hangupsms.cachemanage.handler.SmsTemplateCacheHandler;
import cn.migu.music.hangupsms.cachemanage.handler.subhandler.ReloadAllDbHandle;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionalEventListener;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;

/**
 * @author fgb
 */
@Component
@Slf4j
@RefreshScope
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "q.${hostname}.cache", autoDelete = "true"),
        exchange = @Exchange(value = LocalCacheManager.EXCHANGE_FANOUT_CACHE + "${hangupsms.localcachename}",
                type = ExchangeTypes.FANOUT,
                durable = "true")),
        containerFactory = "cacheRabbitListenerContainerFactory")
public class LocalCacheManager {
    public static final String                           EXCHANGE_FANOUT_CACHE = "exchange.fanout.cache.";

    /**
     * 一个缓存事件对应一个缓存处理器
     */
    private final ConcurrentMap<String, AbsCacheHandler<AbsCacheEvent>> cacheMap = new ConcurrentHashMap<>();

    @Autowired
    private SmsSendTimeCacheHandler smsSendTimeCacheHandler;
    @Autowired
    private SmsSendFrequencyCacheHandler smsSendFrequencyCacheHandler;
    @Autowired
    private SmsSendProvinceCacheHandler smsSendProvinceCacheHandler;
    @Autowired
    private SmsTemplateCacheHandler smsTemplateCacheHandler;
    @Autowired
    private ReloadAllDbHandle reloadAllDbHandle;


    @Value("${hangupsms.localcachename}")
    private String                                       appName;

    @Autowired
    private RabbitTemplate                               rabbitTemplate;

    @Autowired
    private BusinessDbDataCacheServices                  businessDbDataCacheServices;
    /**
     * 执行缓存通知事件线程池
     */
    @Resource(name = "cacheBroadcastExecutor")
    private Executor                                     cacheBroadcastExecutor;

    /**
     * 执行处理事件线程池
     */
    @Resource(name = "cacheEventHandlerExecutor")
    private Executor                                     cacheEventHandlerExecutor;

    /**
     * 广播缓存事件消息
     *
     * @param event
     *            缓存事件
     */
    @TransactionalEventListener(fallbackExecution = true)
    public void broadcastEvent(AbsCacheEvent event) {
        try {
            cacheBroadcastExecutor.execute(new RequestWorker(event));
        } catch (Exception e) {
            LogUtils.warn(e, log, "broadcastEvent failed. event=%s", event);
        }
    }

    @RabbitHandler
    public void cacheEventHandler(AbsCacheEvent event) {
        try {
            cacheEventHandlerExecutor.execute(new ResponseWorker(event));
        } catch (Exception e) {
            LogUtils.warn(e, log, "listener handle cacheEvent failed. event=%s", event);
        }
    }

    /**
     * 获取缓存
     *
     * @param name
     *            缓存名称
     * @return 缓存
     */
    public AbsCacheHandler<AbsCacheEvent> getCache(String name) {
        return cacheMap.get(name);
    }

    /**
     *
     * @param cacheName
     * @param cacheHandler
     */
    public void putCache(String cacheName, AbsCacheHandler<AbsCacheEvent> cacheHandler) {
        cacheMap.putIfAbsent(cacheName, cacheHandler);
    }

    /**
     * 获取所有缓存
     *
     * @return 缓存map
     */
    public ConcurrentMap<String, AbsCacheHandler<AbsCacheEvent>> getCaches() {
        return cacheMap;
    }

    /**
     * 广播缓存事件消息worker
     */
    private final class RequestWorker implements Runnable {
        private final AbsCacheEvent event;

        RequestWorker(AbsCacheEvent event) {
            this.event = event;
        }

        @Override
        public void run() {
            rabbitTemplate.convertAndSend(EXCHANGE_FANOUT_CACHE + appName, "", event);
        }
    }

    /**
     * 处理缓存事件消息worker
     */
    private final class ResponseWorker implements Runnable {
        private final AbsCacheEvent event;

        ResponseWorker(AbsCacheEvent event) {
            this.event = event;
        }

        @Override
        public void run() {
            try {
                LogUtils.info(log, "handle cache event " + event + ",name:" + event.getCacheName());

                switch (event.getCacheName()){
                    case "SMS_TEMPLATE_EVENT":
                        smsTemplateCacheHandler.reload(event);
                        break;
                    case "SMS_SEND_TIME_CHANGED_EVENT":
                        smsSendTimeCacheHandler.reload(event);
                        break;
                    case "SMS_SEND_FREQUENCY_CHANGED_EVENT":
                        smsSendFrequencyCacheHandler.reload(event);
                        break;
                    case "SMS_SEND_PROVINCE_CHANGED_EVENT":
                        smsSendProvinceCacheHandler.reload(event);
                        break;
                    case "ReLoadDBCache":
                        businessDbDataCacheServices.init();
                        break;
                    default:
                        LogUtils.info(log, String.format("未找到句柄处理缓存事件 event=%s, name:%s", event, event.getCacheName()));

                }
            } catch (Exception e) {
                LogUtils.error(e, log, "处理缓存事件消息失败" + event + ",name:" + event.getCacheName());
            }
        }
    }

}
